package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.zhike.bean.TripBean;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/23 15:18
 * @Description: 计算各车各窗口的速度，行驶的里程，急加减次数并写入doris
 */
public class DwsTripWindow {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(2);

        //2.读取car_electricity_log的数据并创建流
        DataStreamSource<String> electricityDS = env.fromSource(KafkaUtils.getKafkaSource("car_electricity_log", "dws_trip_window_1"), WatermarkStrategy.noWatermarks(),"ele_source");

        //转换数据类型
        SingleOutputStreamOperator<JSONObject> eleJsonDS = electricityDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSON.parseObject(value);
            }
        });

        SingleOutputStreamOperator<TripBean> eleTripDS = eleJsonDS.map(new MapFunction<JSONObject, TripBean>() {

            @Override
            public TripBean map(JSONObject value) throws Exception {
                return getFields(value);

            }
        });

        //3.读取car_hybrid_log的数据并创建流
        DataStreamSource<String> hybridDS = env.fromSource(KafkaUtils.getKafkaSource("car_hybrid_log", "dws_trip_window_2"), WatermarkStrategy.noWatermarks(), "hybrid_source");
        SingleOutputStreamOperator<JSONObject> hybridJsonDS = hybridDS.map(JSON::parseObject);

        SingleOutputStreamOperator<TripBean> hybridTripDS = hybridJsonDS.map(new MapFunction<JSONObject, TripBean>() {
            @Override
            public TripBean map(JSONObject value) throws Exception {
                return getFields(value);
            }
        });

        //4.读取car_fuel_log的数据并创建流
        SingleOutputStreamOperator<TripBean> fuelTripDS = env.fromSource(KafkaUtils.getKafkaSource("car_fuel_log", "dws_trip_window_3"), WatermarkStrategy.noWatermarks(), "fuel_source")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TripBean>() {
                    @Override
                    public TripBean map(JSONObject value) throws Exception {
                        return getFields(value);
                    }
                });

        //5.合并三个流
        DataStream<TripBean> unionDS = eleTripDS.union(hybridTripDS)
                .union(fuelTripDS);

        //6.设置水位线
        SingleOutputStreamOperator<TripBean> tripBeanSingleOutputStreamOperator = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TripBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TripBean>() {
                            @Override
                            public long extractTimestamp(TripBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );
        
        //7.按照vin分组
        KeyedStream<TripBean, String> tripBeanStringKeyedStream = tripBeanSingleOutputStreamOperator.keyBy(new KeySelector<TripBean, String>() {
            @Override
            public String getKey(TripBean value) throws Exception {
                return value.getVin();
            }
        });

        //8.通过状态编程计算是否存在急加减次数
        SingleOutputStreamOperator<TripBean> velocityStateDS = tripBeanStringKeyedStream.map(new RichMapFunction<TripBean, TripBean>() {
            private ValueState<Double> velocityState;

            @Override
            public void open(Configuration parameters) throws Exception {
                velocityState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("velocity_state", Double.class));
            }

            @Override
            public TripBean map(TripBean value) throws Exception {
                //获取当前速度
                Double totalSpeed = value.getTotalSpeed();
                //获取状态中的值
                Double velocityPre = velocityState.value();
                if (velocityPre == null) {
                    velocityState.update(totalSpeed);
                } else {
                    if (Math.abs(totalSpeed - velocityPre) > 70) {
                        value.setDangerCount(value.getDangerCount() + 1);
                        velocityState.update(totalSpeed);
                    }
                    velocityState.update(totalSpeed);
                }
                return value;
            }
        });



        //9.开窗并聚合
        SingleOutputStreamOperator<TripBean> resultDS =
                velocityStateDS.keyBy(new KeySelector<TripBean, String>() {
                            @Override
                            public String getKey(TripBean value) throws Exception {
                                return value.getVin();
                            }
                        })
        .window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<TripBean>() {
                            @Override
                            public TripBean reduce(TripBean value1, TripBean value2) throws Exception {
                                value1.setMaxMileage(Math.max(value1.getMaxMileage(), value2.getMaxMileage()));
                                value1.setMinMileage(Math.min(value1.getMinMileage(), value2.getMinMileage()));
                                value1.setTotalSpeed(value1.getTotalSpeed() + value2.getTotalSpeed());
                                value1.setRunningCount(value1.getRunningCount() + value2.getRunningCount());
                                value1.setDangerCount(value1.getDangerCount() + value2.getDangerCount());
                                return value1;
                            }
                        }, new WindowFunction<TripBean, TripBean,String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<TripBean> input, Collector<TripBean> out) throws Exception {
                                TripBean tripBean = input.iterator().next();
                                tripBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                tripBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                                tripBean.setTs(System.currentTimeMillis());
                                tripBean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                                out.collect(tripBean);
                            }
                        });


        resultDS.print();

        //10.将结果写入doris表
        SingleOutputStreamOperator<String> result = resultDS.map(new MapFunction<TripBean, String>() {
            @Override
            public String map(TripBean tripBean) throws Exception {
                SerializeConfig serializeConfig = new SerializeConfig();
                serializeConfig.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                return JSONObject.toJSONString(tripBean, serializeConfig, SerializerFeature.WriteMapNullValue);
            }
        });
        result.sinkTo(DorisUtils.getDorisSink("dws_trip_window"));


        //11.启动任务
        try {
            env.execute("DwsTripWindow");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }

    private static TripBean getFields(JSONObject value) {
        //获取汽车id
        String vin = value.getString("vin");
        //获取里程
        double mileage = value.getInteger("mileage") / 100.0;
        //获取当前速度（分子）
        double velocity = value.getInteger("velocity") / 10.0;
        //计算时速分母
        Integer velocityCount = velocity > 0 ? 1 : 0;
        //急加减次数
        Integer dangerCount = 0;
        //获取时间戳
        Long ts = value.getLong("timestamp");
        return TripBean.builder()
                .vin(vin)
                .maxMileage(mileage)
                .minMileage(mileage)
                .totalSpeed(velocity)
                .dangerCount(dangerCount)
                .runningCount(velocityCount)
                .ts(ts)
                .build();
    }
}
